/*
 * Licensed to the Tai Ping Jin Ke
 *
 * Copyright (c) 2022 .
 * All rights reserved.
 * 项目名称：太平金科-基础组件封装层-分布式协调工具Zookeeper-缺省实现
 * 版权说明：本软件属太平金融科技服务(上海)有限公司所有，在未获得太平金融科技服务(上海)有限公司正式授权情况下，任何企业和个人，不能获取、阅读、安装、传播本软件涉及的任何受知识产权保护的内容。
 */
package com.taipingframework.boot.zookeeper.support;

import com.taipingframework.boot.zookeeper.properties.ZookeeperProperty;
import com.taipingframework.utility.context.SpringContextHolder;
import com.taipingframework.utility.extend.ExceptionHelper;
import lombok.extern.slf4j.Slf4j;
import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Zookeeper临时节点的生命周期就是Zookeeper Session会话的生命周期，会话结束自动删除
 */
@Slf4j
public class ZookeeperLock {

    // zk分布式锁存放路径
    private static final String ZK_LOCK_PATH = "/lock";

    private static ZookeeperProperty zkProp;

    // zk客户端工具
    private ZkClient zkClient;

    // 事件监听对象
    private IZkDataListener zkDataListener = new ZkLockListener();

    // 信号量
    private CountDownLatch countDownLatch;

    // 计数器
    private AtomicInteger counter = new AtomicInteger(0);

    static {
        zkProp = SpringContextHolder.getBean(ZookeeperProperty.class);
    }

    public ZookeeperLock() {
        this.zkClient = new ZkClient(zkProp.getConnectString(), zkProp.getSessionTimeout());
    }

    /**
     * 获取锁
     */
    public void gainLock() {
        if (this.tryGainLock()) {
            log.info("获取锁 - 尝试次数：{}", counter.get());
            return;
        }
        this.waitForLock();
        this.gainLock();
    }

    // 尝试获取锁，成功-true; 失败-false
    private boolean tryGainLock() {
        try {
            counter.incrementAndGet();
            zkClient.createEphemeral(ZK_LOCK_PATH);
            return true;
        } catch (Exception ignored) {
            return false;
        }
    }

    // 等待锁释放
    private void waitForLock() {
        // 注册节点信息 - 事件监听
        zkClient.subscribeDataChanges(ZK_LOCK_PATH, zkDataListener);

        // 等待锁释放
        if (zkClient.exists(ZK_LOCK_PATH)) {
            countDownLatch = new CountDownLatch(1);
            try {
                countDownLatch.await(); //① 阻塞(等待节点被删除)
            } catch (Exception ex) {
                log.error(ExceptionHelper.printException(ex));
            }
        }

        // 删除节点信息 - 事件监听
        zkClient.unsubscribeDataChanges(ZK_LOCK_PATH, zkDataListener);
    }

    /**
     * 释放锁
     */
    public void releaseLock() {
        log.info("释放锁");
        if (zkClient != null) {
            zkClient.close(); //③ 结束Zookeeper Session会话
        }
    }

    @Override
    protected void finalize() throws Throwable {
        if (zkClient != null) {
            zkClient.close();
        }
        super.finalize();
    }

    class ZkLockListener implements IZkDataListener {
        @Override
        public void handleDataChange(String dataPath, Object data) {
            // 节点Change事件
        }

        @Override
        public void handleDataDeleted(String dataPath) {
            // 节点Delete事件
            if (countDownLatch != null) {
                countDownLatch.countDown(); //② 解除阻塞
            }
        }
    }
}
